package com.jeffse.light.stream.core.data.operator.Covert;

import com.jeffse.light.stream.core.data.descriptor.operatorDescriptor.OperatorDescriptor;
import com.jeffse.light.stream.core.function.advance.FlatMapFunction;

public class FlatMapConvert<T> implements BaseConvert<T> {
    @Override
    public void convert(OperatorDescriptor operatorDescriptor, T value) throws Exception {
        FlatMapFunction flatMapFunction = (FlatMapFunction) operatorDescriptor.getFunc();
        flatMapFunction.flatMap(value, (record)-> operatorDescriptor.getCtx().collect(record, operatorDescriptor.getNetOperatorDescriptor(), operatorDescriptor.getOutTag()));
    }
}
